nanopyx.liquid.__liquid_engine__
1import os 2import timeit 3import yaml 4import datetime 5import inspect 6from functools import partial 7from itertools import combinations 8from pathlib import Path 9 10import numpy as np 11 12# This will in the future come from the Agent 13from .__njit__ import njit_works 14from .__opencl__ import opencl_works, devices 15 16__home_folder__ = os.path.expanduser("~") 17__benchmark_folder__ = os.path.join(__home_folder__, ".nanopyx") 18if not os.path.exists(__benchmark_folder__): 19 os.makedirs(__benchmark_folder__) 20 21class LiquidEngine: 22 23 """ 24 Base class for parts of the Nanopyx Liquid Engine 25 Vroom Vroom 26 """ 27 28 def __init__(self, testing:bool=False, 29 opencl_:bool = False, unthreaded_:bool = False, 30 threaded_:bool = False, threaded_static_:bool = False, 31 threaded_dynamic_:bool = False, threaded_guided_:bool = False, 32 python_:bool=False, njit_:bool=False, clear_benchmarks:bool=False) -> None: 33 """ 34 Initialize the Liquid Engine 35 The Liquid Engine base class is inherited by children classes that implement specific methods 36 37 Engine responsabilities: 38 1. Store implemented run types; 39 2. Handle previous benchmarks and I/O; 40 2. When queried, benchmark all available run types; 41 3. Run a specific method using a selected run type; 42 43 Benchmark files have the following format: 44 The benchmark file is read as dict of dicts. 45 BENCHMARK DICT FOR A SPECIFIC METHOD 46 |- RUN_TYPE #1 47 | |- ARGS_REPR #1 48 | | |- [score, t2run#1, t2run#2, t2run#3, ...] last are newer. nan means fail 49 | |- ARGS_REPR #2 50 | | |- [score, t2run#1, t2run#2, t2run#3, ...] last are newer. nan means fail 51 | (...) 52 |- RUN_TYPE #2 53 (...) 54 """ 55 56 # Start by checking available run types 57 self._run_types = {} 58 if opencl_ and opencl_works(): 59 for d in devices: 60 self._run_types[f"OpenCL_{d['device'].name}"] = partial(self._run_opencl, device=d) 61 if threaded_: 62 self._run_types["Threaded"] = self._run_threaded 63 if unthreaded_: 64 self._run_types["Unthreaded"] = self._run_unthreaded 65 if threaded_static_: 66 self._run_types["Threaded_static"] = self._run_threaded_static 67 if threaded_dynamic_: 68 self._run_types["Threaded_dynamic"] = self._run_threaded_dynamic 69 if threaded_guided_: 70 self._run_types["Threaded_guided"] = self._run_threaded_guided 71 if python_: 72 self._run_types["Python"] = self._run_python 73 if njit_ and njit_works(): 74 self._run_types["Numba"] = self._run_njit 75 # Try to trigger early compilation 76 try: 77 self._run_njit() 78 except TypeError: 79 print("Consider adding default arguments to the njit implementation to trigger early compilation") 80 81 self.testing = testing 82 83 # benchmarks file path 84 # e.g.: ~/.nanopyx/liquid/_le_interpolation_nearest_neighbor.cpython-310-darwin/ShiftAndMagnify.yml 85 base_path = os.path.join(__benchmark_folder__,"liquid", 86 os.path.split(os.path.splitext(inspect.getfile(self.__class__))[0])[1]) 87 os.makedirs(base_path, exist_ok=True) 88 self._benchmark_filepath = os.path.join(base_path,self.__class__.__name__+".yml") 89 90 # Load config file if it exists, otherwise create an empty config 91 if not clear_benchmarks and os.path.exists(self._benchmark_filepath): 92 with open(self._benchmark_filepath) as f: 93 self._benchmarks = yaml.load(f, Loader=yaml.FullLoader) 94 else: 95 self._benchmarks = {} 96 97 # check if the cfg dictionary has a key for every available run type 98 for run_type_designation in self._run_types.keys(): 99 if run_type_designation not in self._benchmarks: 100 self._benchmarks[run_type_designation] = {} 101 102 # helper attribute for benchmarking function 103 self._last_args = None 104 self._last_runtype = None 105 self._last_time = None 106 107 def _run(self, *args, run_type:str, **kwargs): 108 """ 109 Runs the function with the given args and kwargs 110 111 The code above does the following: 112 1. Check the specified run_type 113 - if str checks if the run type exists otherwise raise a NotImplementedError 114 2. It will run the _run_{run_type} function 115 3. It will return the result and the time taken to run 116 117 :param args: args for the function 118 :param run_type: the run type to use 119 :param kwargs: kwargs for the function 120 :return: the result and time taken 121 """ 122 123 if run_type not in self._run_types: 124 print(f"Unexpected run type {run_type}") 125 raise NotImplementedError 126 127 # try to run 128 try: 129 t_start = timeit.default_timer() 130 result = self._run_types[run_type](*args, **kwargs) 131 t2run = timeit.default_timer()-t_start 132 except Exception as e: 133 print(f"Unexpected error while trying to run {run_type}") 134 print(e) 135 print("Please try again with another run type") 136 result = None 137 t2run = np.inf 138 139 arg_repr, arg_score = self._get_args_repr_score(*args, **kwargs) 140 self._store_results(arg_repr, arg_score, run_type, t2run) 141 142 self._last_time = t2run 143 self._last_args = arg_repr 144 self._last_runtype = run_type 145 146 return result 147 148 def benchmark(self,*args, **kwargs): 149 """ 150 1. Run each available run type and record the run time and return value 151 2. Sort the run times from fastest to slowest 152 3. Compare each run type against each other, sorted by speed 153 154 :param args: args for the run method 155 :param kwargs: kwargs for the run method 156 :return: a list of tuples containing the run time, run type name and optionally the return values 157 :rtype: [[run_time, run_type_name, return_value], ...] 158 """ 159 160 # Create some lists to store runtimes and return values of run types 161 run_times = {} 162 returns = {} 163 164 # Run each run type and record the run time and return value 165 for run_type in self._run_types: 166 167 r = self._run(*args, run_type=run_type, **kwargs) 168 169 run_times[run_type] = self._last_time 170 171 if self.testing: # Store return values if testing 172 returns[run_type] = r 173 else: 174 returns[run_type] = None 175 176 # Sort run_times by value 177 speed_sort = [] 178 for run_type in sorted(run_times, key=run_times.get, reverse=False): 179 speed_sort.append( 180 ( 181 run_times[run_type], 182 run_type, 183 returns[run_type], 184 ) 185 ) 186 187 print(f"Fastest run type: {speed_sort[0][1]}") 188 print(f"Slowest run type: {speed_sort[-1][1]}") 189 190 # Compare each run type against each other, sorted by speed 191 for pair in combinations(speed_sort,2): 192 193 print(f"{pair[0][1]} is {pair[1][0]/pair[0][0]:.2f}x faster than {pair[1][1]}") 194 if self.testing: 195 if self._compare_runs(pair[0][2],pair[1][2]): 196 print(f"{pair[0][1]} and {pair[1][1]} have similar outputs!") 197 else: 198 print(f"WARNING: outputs of {pair[0][1]} and {pair[1][1]} don't match!") 199 200 return speed_sort 201 202 def _compare_runs(self, output_1, output_2): 203 204 mse = np.average((output_1 - output_2)**2) 205 206 if mse < 0.1: 207 return True 208 else: 209 return False 210 211 212 def _get_cl_code(self, file_name, cl_dp): 213 """ 214 Retrieves the OpenCL code from the corresponding .cl file 215 """ 216 cl_file = os.path.splitext(file_name)[0] + ".cl" 217 if not os.path.exists(cl_file): 218 cl_file = Path(__file__).parent / file_name 219 220 assert os.path.exists(cl_file), "Could not find OpenCL file: " + cl_file 221 222 kernel_str = open(cl_file).read() 223 224 if not cl_dp: 225 kernel_str = kernel_str.replace("double", "float") 226 227 return kernel_str 228 229 def _store_results(self, arg_repr, arg_score, run_type, t2run): 230 """ 231 Stores the results of a run 232 """ 233 234 # Re-read the benchmark file in case it has been updated 235 try: 236 with open(self._benchmark_filepath) as f: 237 self._benchmarks = yaml.load(f, Loader=yaml.FullLoader) 238 except FileNotFoundError: 239 self._benchmarks = self._benchmarks 240 241 # Check if the run type has been run, and if not create empty info 242 run_type_benchs = self._benchmarks[run_type] 243 if arg_repr not in run_type_benchs: 244 run_type_benchs[arg_repr] = [arg_score] 245 246 # Get the run info 247 c = run_type_benchs[arg_repr] 248 249 assert c[0] == arg_score, "arg_score mismatch" 250 251 # if run failed, t2run is np.inf 252 if np.isinf(t2run): 253 c.append(np.nan) 254 else: 255 c.append(t2run) 256 257 self._dump_run_times() 258 259 def _dump_run_times(self,): 260 """We might need to wrap this into a multiprocessing.Queue if we find it blocking""" 261 with open(self._benchmark_filepath, "w") as f: 262 yaml.dump(self._benchmarks, f) 263 264 def _get_args_repr_score(self, *args, **kwargs): 265 """ 266 Get a string representation of the args and kwargs and corresponding 'score' / 'norm' 267 The idea is that similar args have closer 'score'. Fuzzy logic 268 269 The code does the following: 270 1. It converts any args that are floats or ints to "number()" strings, and any args that are tensors to "shape()" strings 271 2. It converts any kwargs that are floats or ints to "number()" strings, and any kwargs that are tensors to "shape()" strings 272 3. The 'score' is given by the product of all the floats or ints and all the shape sizes. 273 274 :return: the string representation of the args and kwargs 275 :rtype: str 276 """ 277 _norm = 1 278 _args = [] 279 for arg in args: 280 if type(arg) in (float, int): 281 _args.append(f"number({arg})") 282 if arg==0: 283 arg=1 284 _norm *= arg 285 elif hasattr(arg, "shape"): 286 _args.append(f"shape{arg.shape}") 287 _norm *= arg.size 288 else: 289 _args.append(arg) 290 291 _kwargs = {} 292 for k, v in kwargs.items(): 293 if type(v) in (float, int): 294 _kwargs[k] = f"number({v})" 295 if v==0: 296 v=1 297 _norm *= v 298 if hasattr(v, "shape"): 299 _kwargs[k] = f"shape{arg.shape}" 300 _norm *= v.size 301 else: 302 _kwargs[k] = v 303 304 return repr((_args, _kwargs)), _norm 305 306 def get_highest_divisor(self, size_, max_): 307 """ 308 Returns the highest divisor of size_ that is still lower than max_ 309 """ 310 value = 1 311 for i in range(1, int(np.sqrt(size_)+1)): 312 if size_ % i==0: 313 if i*i != size_: 314 div2 = size_/i 315 316 if i < max_: 317 value = max(value, i) 318 if div2 < max_: 319 value = max(value, div2) 320 return int(value) 321 322 def get_work_group(self, device, shape): 323 """ 324 Calculates work group size for a given device and shape of global work space 325 """ 326 327 max_wg_dims = device.max_work_item_sizes[0:3] 328 max_glo_dims = device.max_work_group_size 329 330 three = self.get_highest_divisor(shape[2], max_wg_dims[2]) 331 max_two = max_glo_dims/three 332 two = self.get_highest_divisor(shape[1], max_two) 333 one = 1 334 #return (1,1,1) 335 return (one, two, three) 336 337 338 339 ##################################################### 340 # RUN METHODS # 341 # THESE SHOULD ALWAYS BE OVERRIDEN BY CHILD CLASSES # 342 ##################################################### 343 344 def run(self, *args, **kwargs): 345 """ 346 Runs the function with the given args and kwargs 347 Should be overridden by the any class that inherits from this class 348 """ 349 return self._run(*args, **kwargs) 350 351 def _run_opencl(*args, **kwargs): 352 """ 353 Runs the OpenCL version of the function 354 Should be overridden by the any class that inherits from this class 355 """ 356 pass 357 358 def _run_unthreaded(*args, **kwargs): 359 """ 360 Runs the cython unthreaded version of the function 361 Should be overridden by the any class that inherits from this class 362 """ 363 pass 364 365 def _run_threaded(*args, **kwargs): 366 """ 367 Runs the cython threaded version of the function 368 Should be overridden by the any class that inherits from this class 369 """ 370 pass 371 372 def _run_threaded_static(*args, **kwargs): 373 """ 374 Runs the cython threaded static version of the function 375 Should be overridden by the any class that inherits from this class 376 """ 377 pass 378 379 def _run_threaded_dynamic(*args, **kwargs): 380 """ 381 Runs the cython threaded dynamic version of the function 382 Should be overridden by the any class that inherits from this class 383 """ 384 pass 385 386 def _run_threaded_guided(*args, **kwargs): 387 """ 388 Runs the cython threaded guided version of the function 389 Should be overridden by the any class that inherits from this class 390 """ 391 pass 392 393 def _run_python(*args, **kwargs): 394 """ 395 Runs the python version of the function 396 Should be overridden by the any class that inherits from this class 397 """ 398 pass 399 400 def _run_njit(*args, **kwargs): 401 """ 402 Runs the njit version of the function 403 Should be overridden by the any class that inherits from this class 404 """ 405 pass
class
LiquidEngine:
22class LiquidEngine: 23 24 """ 25 Base class for parts of the Nanopyx Liquid Engine 26 Vroom Vroom 27 """ 28 29 def __init__(self, testing:bool=False, 30 opencl_:bool = False, unthreaded_:bool = False, 31 threaded_:bool = False, threaded_static_:bool = False, 32 threaded_dynamic_:bool = False, threaded_guided_:bool = False, 33 python_:bool=False, njit_:bool=False, clear_benchmarks:bool=False) -> None: 34 """ 35 Initialize the Liquid Engine 36 The Liquid Engine base class is inherited by children classes that implement specific methods 37 38 Engine responsabilities: 39 1. Store implemented run types; 40 2. Handle previous benchmarks and I/O; 41 2. When queried, benchmark all available run types; 42 3. Run a specific method using a selected run type; 43 44 Benchmark files have the following format: 45 The benchmark file is read as dict of dicts. 46 BENCHMARK DICT FOR A SPECIFIC METHOD 47 |- RUN_TYPE #1 48 | |- ARGS_REPR #1 49 | | |- [score, t2run#1, t2run#2, t2run#3, ...] last are newer. nan means fail 50 | |- ARGS_REPR #2 51 | | |- [score, t2run#1, t2run#2, t2run#3, ...] last are newer. nan means fail 52 | (...) 53 |- RUN_TYPE #2 54 (...) 55 """ 56 57 # Start by checking available run types 58 self._run_types = {} 59 if opencl_ and opencl_works(): 60 for d in devices: 61 self._run_types[f"OpenCL_{d['device'].name}"] = partial(self._run_opencl, device=d) 62 if threaded_: 63 self._run_types["Threaded"] = self._run_threaded 64 if unthreaded_: 65 self._run_types["Unthreaded"] = self._run_unthreaded 66 if threaded_static_: 67 self._run_types["Threaded_static"] = self._run_threaded_static 68 if threaded_dynamic_: 69 self._run_types["Threaded_dynamic"] = self._run_threaded_dynamic 70 if threaded_guided_: 71 self._run_types["Threaded_guided"] = self._run_threaded_guided 72 if python_: 73 self._run_types["Python"] = self._run_python 74 if njit_ and njit_works(): 75 self._run_types["Numba"] = self._run_njit 76 # Try to trigger early compilation 77 try: 78 self._run_njit() 79 except TypeError: 80 print("Consider adding default arguments to the njit implementation to trigger early compilation") 81 82 self.testing = testing 83 84 # benchmarks file path 85 # e.g.: ~/.nanopyx/liquid/_le_interpolation_nearest_neighbor.cpython-310-darwin/ShiftAndMagnify.yml 86 base_path = os.path.join(__benchmark_folder__,"liquid", 87 os.path.split(os.path.splitext(inspect.getfile(self.__class__))[0])[1]) 88 os.makedirs(base_path, exist_ok=True) 89 self._benchmark_filepath = os.path.join(base_path,self.__class__.__name__+".yml") 90 91 # Load config file if it exists, otherwise create an empty config 92 if not clear_benchmarks and os.path.exists(self._benchmark_filepath): 93 with open(self._benchmark_filepath) as f: 94 self._benchmarks = yaml.load(f, Loader=yaml.FullLoader) 95 else: 96 self._benchmarks = {} 97 98 # check if the cfg dictionary has a key for every available run type 99 for run_type_designation in self._run_types.keys(): 100 if run_type_designation not in self._benchmarks: 101 self._benchmarks[run_type_designation] = {} 102 103 # helper attribute for benchmarking function 104 self._last_args = None 105 self._last_runtype = None 106 self._last_time = None 107 108 def _run(self, *args, run_type:str, **kwargs): 109 """ 110 Runs the function with the given args and kwargs 111 112 The code above does the following: 113 1. Check the specified run_type 114 - if str checks if the run type exists otherwise raise a NotImplementedError 115 2. It will run the _run_{run_type} function 116 3. It will return the result and the time taken to run 117 118 :param args: args for the function 119 :param run_type: the run type to use 120 :param kwargs: kwargs for the function 121 :return: the result and time taken 122 """ 123 124 if run_type not in self._run_types: 125 print(f"Unexpected run type {run_type}") 126 raise NotImplementedError 127 128 # try to run 129 try: 130 t_start = timeit.default_timer() 131 result = self._run_types[run_type](*args, **kwargs) 132 t2run = timeit.default_timer()-t_start 133 except Exception as e: 134 print(f"Unexpected error while trying to run {run_type}") 135 print(e) 136 print("Please try again with another run type") 137 result = None 138 t2run = np.inf 139 140 arg_repr, arg_score = self._get_args_repr_score(*args, **kwargs) 141 self._store_results(arg_repr, arg_score, run_type, t2run) 142 143 self._last_time = t2run 144 self._last_args = arg_repr 145 self._last_runtype = run_type 146 147 return result 148 149 def benchmark(self,*args, **kwargs): 150 """ 151 1. Run each available run type and record the run time and return value 152 2. Sort the run times from fastest to slowest 153 3. Compare each run type against each other, sorted by speed 154 155 :param args: args for the run method 156 :param kwargs: kwargs for the run method 157 :return: a list of tuples containing the run time, run type name and optionally the return values 158 :rtype: [[run_time, run_type_name, return_value], ...] 159 """ 160 161 # Create some lists to store runtimes and return values of run types 162 run_times = {} 163 returns = {} 164 165 # Run each run type and record the run time and return value 166 for run_type in self._run_types: 167 168 r = self._run(*args, run_type=run_type, **kwargs) 169 170 run_times[run_type] = self._last_time 171 172 if self.testing: # Store return values if testing 173 returns[run_type] = r 174 else: 175 returns[run_type] = None 176 177 # Sort run_times by value 178 speed_sort = [] 179 for run_type in sorted(run_times, key=run_times.get, reverse=False): 180 speed_sort.append( 181 ( 182 run_times[run_type], 183 run_type, 184 returns[run_type], 185 ) 186 ) 187 188 print(f"Fastest run type: {speed_sort[0][1]}") 189 print(f"Slowest run type: {speed_sort[-1][1]}") 190 191 # Compare each run type against each other, sorted by speed 192 for pair in combinations(speed_sort,2): 193 194 print(f"{pair[0][1]} is {pair[1][0]/pair[0][0]:.2f}x faster than {pair[1][1]}") 195 if self.testing: 196 if self._compare_runs(pair[0][2],pair[1][2]): 197 print(f"{pair[0][1]} and {pair[1][1]} have similar outputs!") 198 else: 199 print(f"WARNING: outputs of {pair[0][1]} and {pair[1][1]} don't match!") 200 201 return speed_sort 202 203 def _compare_runs(self, output_1, output_2): 204 205 mse = np.average((output_1 - output_2)**2) 206 207 if mse < 0.1: 208 return True 209 else: 210 return False 211 212 213 def _get_cl_code(self, file_name, cl_dp): 214 """ 215 Retrieves the OpenCL code from the corresponding .cl file 216 """ 217 cl_file = os.path.splitext(file_name)[0] + ".cl" 218 if not os.path.exists(cl_file): 219 cl_file = Path(__file__).parent / file_name 220 221 assert os.path.exists(cl_file), "Could not find OpenCL file: " + cl_file 222 223 kernel_str = open(cl_file).read() 224 225 if not cl_dp: 226 kernel_str = kernel_str.replace("double", "float") 227 228 return kernel_str 229 230 def _store_results(self, arg_repr, arg_score, run_type, t2run): 231 """ 232 Stores the results of a run 233 """ 234 235 # Re-read the benchmark file in case it has been updated 236 try: 237 with open(self._benchmark_filepath) as f: 238 self._benchmarks = yaml.load(f, Loader=yaml.FullLoader) 239 except FileNotFoundError: 240 self._benchmarks = self._benchmarks 241 242 # Check if the run type has been run, and if not create empty info 243 run_type_benchs = self._benchmarks[run_type] 244 if arg_repr not in run_type_benchs: 245 run_type_benchs[arg_repr] = [arg_score] 246 247 # Get the run info 248 c = run_type_benchs[arg_repr] 249 250 assert c[0] == arg_score, "arg_score mismatch" 251 252 # if run failed, t2run is np.inf 253 if np.isinf(t2run): 254 c.append(np.nan) 255 else: 256 c.append(t2run) 257 258 self._dump_run_times() 259 260 def _dump_run_times(self,): 261 """We might need to wrap this into a multiprocessing.Queue if we find it blocking""" 262 with open(self._benchmark_filepath, "w") as f: 263 yaml.dump(self._benchmarks, f) 264 265 def _get_args_repr_score(self, *args, **kwargs): 266 """ 267 Get a string representation of the args and kwargs and corresponding 'score' / 'norm' 268 The idea is that similar args have closer 'score'. Fuzzy logic 269 270 The code does the following: 271 1. It converts any args that are floats or ints to "number()" strings, and any args that are tensors to "shape()" strings 272 2. It converts any kwargs that are floats or ints to "number()" strings, and any kwargs that are tensors to "shape()" strings 273 3. The 'score' is given by the product of all the floats or ints and all the shape sizes. 274 275 :return: the string representation of the args and kwargs 276 :rtype: str 277 """ 278 _norm = 1 279 _args = [] 280 for arg in args: 281 if type(arg) in (float, int): 282 _args.append(f"number({arg})") 283 if arg==0: 284 arg=1 285 _norm *= arg 286 elif hasattr(arg, "shape"): 287 _args.append(f"shape{arg.shape}") 288 _norm *= arg.size 289 else: 290 _args.append(arg) 291 292 _kwargs = {} 293 for k, v in kwargs.items(): 294 if type(v) in (float, int): 295 _kwargs[k] = f"number({v})" 296 if v==0: 297 v=1 298 _norm *= v 299 if hasattr(v, "shape"): 300 _kwargs[k] = f"shape{arg.shape}" 301 _norm *= v.size 302 else: 303 _kwargs[k] = v 304 305 return repr((_args, _kwargs)), _norm 306 307 def get_highest_divisor(self, size_, max_): 308 """ 309 Returns the highest divisor of size_ that is still lower than max_ 310 """ 311 value = 1 312 for i in range(1, int(np.sqrt(size_)+1)): 313 if size_ % i==0: 314 if i*i != size_: 315 div2 = size_/i 316 317 if i < max_: 318 value = max(value, i) 319 if div2 < max_: 320 value = max(value, div2) 321 return int(value) 322 323 def get_work_group(self, device, shape): 324 """ 325 Calculates work group size for a given device and shape of global work space 326 """ 327 328 max_wg_dims = device.max_work_item_sizes[0:3] 329 max_glo_dims = device.max_work_group_size 330 331 three = self.get_highest_divisor(shape[2], max_wg_dims[2]) 332 max_two = max_glo_dims/three 333 two = self.get_highest_divisor(shape[1], max_two) 334 one = 1 335 #return (1,1,1) 336 return (one, two, three) 337 338 339 340 ##################################################### 341 # RUN METHODS # 342 # THESE SHOULD ALWAYS BE OVERRIDEN BY CHILD CLASSES # 343 ##################################################### 344 345 def run(self, *args, **kwargs): 346 """ 347 Runs the function with the given args and kwargs 348 Should be overridden by the any class that inherits from this class 349 """ 350 return self._run(*args, **kwargs) 351 352 def _run_opencl(*args, **kwargs): 353 """ 354 Runs the OpenCL version of the function 355 Should be overridden by the any class that inherits from this class 356 """ 357 pass 358 359 def _run_unthreaded(*args, **kwargs): 360 """ 361 Runs the cython unthreaded version of the function 362 Should be overridden by the any class that inherits from this class 363 """ 364 pass 365 366 def _run_threaded(*args, **kwargs): 367 """ 368 Runs the cython threaded version of the function 369 Should be overridden by the any class that inherits from this class 370 """ 371 pass 372 373 def _run_threaded_static(*args, **kwargs): 374 """ 375 Runs the cython threaded static version of the function 376 Should be overridden by the any class that inherits from this class 377 """ 378 pass 379 380 def _run_threaded_dynamic(*args, **kwargs): 381 """ 382 Runs the cython threaded dynamic version of the function 383 Should be overridden by the any class that inherits from this class 384 """ 385 pass 386 387 def _run_threaded_guided(*args, **kwargs): 388 """ 389 Runs the cython threaded guided version of the function 390 Should be overridden by the any class that inherits from this class 391 """ 392 pass 393 394 def _run_python(*args, **kwargs): 395 """ 396 Runs the python version of the function 397 Should be overridden by the any class that inherits from this class 398 """ 399 pass 400 401 def _run_njit(*args, **kwargs): 402 """ 403 Runs the njit version of the function 404 Should be overridden by the any class that inherits from this class 405 """ 406 pass
Base class for parts of the Nanopyx Liquid Engine Vroom Vroom
LiquidEngine( testing: bool = False, opencl_: bool = False, unthreaded_: bool = False, threaded_: bool = False, threaded_static_: bool = False, threaded_dynamic_: bool = False, threaded_guided_: bool = False, python_: bool = False, njit_: bool = False, clear_benchmarks: bool = False)
29 def __init__(self, testing:bool=False, 30 opencl_:bool = False, unthreaded_:bool = False, 31 threaded_:bool = False, threaded_static_:bool = False, 32 threaded_dynamic_:bool = False, threaded_guided_:bool = False, 33 python_:bool=False, njit_:bool=False, clear_benchmarks:bool=False) -> None: 34 """ 35 Initialize the Liquid Engine 36 The Liquid Engine base class is inherited by children classes that implement specific methods 37 38 Engine responsabilities: 39 1. Store implemented run types; 40 2. Handle previous benchmarks and I/O; 41 2. When queried, benchmark all available run types; 42 3. Run a specific method using a selected run type; 43 44 Benchmark files have the following format: 45 The benchmark file is read as dict of dicts. 46 BENCHMARK DICT FOR A SPECIFIC METHOD 47 |- RUN_TYPE #1 48 | |- ARGS_REPR #1 49 | | |- [score, t2run#1, t2run#2, t2run#3, ...] last are newer. nan means fail 50 | |- ARGS_REPR #2 51 | | |- [score, t2run#1, t2run#2, t2run#3, ...] last are newer. nan means fail 52 | (...) 53 |- RUN_TYPE #2 54 (...) 55 """ 56 57 # Start by checking available run types 58 self._run_types = {} 59 if opencl_ and opencl_works(): 60 for d in devices: 61 self._run_types[f"OpenCL_{d['device'].name}"] = partial(self._run_opencl, device=d) 62 if threaded_: 63 self._run_types["Threaded"] = self._run_threaded 64 if unthreaded_: 65 self._run_types["Unthreaded"] = self._run_unthreaded 66 if threaded_static_: 67 self._run_types["Threaded_static"] = self._run_threaded_static 68 if threaded_dynamic_: 69 self._run_types["Threaded_dynamic"] = self._run_threaded_dynamic 70 if threaded_guided_: 71 self._run_types["Threaded_guided"] = self._run_threaded_guided 72 if python_: 73 self._run_types["Python"] = self._run_python 74 if njit_ and njit_works(): 75 self._run_types["Numba"] = self._run_njit 76 # Try to trigger early compilation 77 try: 78 self._run_njit() 79 except TypeError: 80 print("Consider adding default arguments to the njit implementation to trigger early compilation") 81 82 self.testing = testing 83 84 # benchmarks file path 85 # e.g.: ~/.nanopyx/liquid/_le_interpolation_nearest_neighbor.cpython-310-darwin/ShiftAndMagnify.yml 86 base_path = os.path.join(__benchmark_folder__,"liquid", 87 os.path.split(os.path.splitext(inspect.getfile(self.__class__))[0])[1]) 88 os.makedirs(base_path, exist_ok=True) 89 self._benchmark_filepath = os.path.join(base_path,self.__class__.__name__+".yml") 90 91 # Load config file if it exists, otherwise create an empty config 92 if not clear_benchmarks and os.path.exists(self._benchmark_filepath): 93 with open(self._benchmark_filepath) as f: 94 self._benchmarks = yaml.load(f, Loader=yaml.FullLoader) 95 else: 96 self._benchmarks = {} 97 98 # check if the cfg dictionary has a key for every available run type 99 for run_type_designation in self._run_types.keys(): 100 if run_type_designation not in self._benchmarks: 101 self._benchmarks[run_type_designation] = {} 102 103 # helper attribute for benchmarking function 104 self._last_args = None 105 self._last_runtype = None 106 self._last_time = None
Initialize the Liquid Engine The Liquid Engine base class is inherited by children classes that implement specific methods
Engine responsabilities:
- Store implemented run types;
- Handle previous benchmarks and I/O;
- When queried, benchmark all available run types;
- Run a specific method using a selected run type;
Benchmark files have the following format:
The benchmark file is read as dict of dicts.
BENCHMARK DICT FOR A SPECIFIC METHOD
|- RUN_TYPE #1
| |- ARGS_REPR #1
| | |- [score, t2run#1, t2run#2, t2run#3, ...] last are newer. nan means fail
| |- ARGS_REPR #2
| | |- [score, t2run#1, t2run#2, t2run#3, ...] last are newer. nan means fail
| (...)
|- RUN_TYPE #2
(...)
def
benchmark(self, *args, **kwargs):
149 def benchmark(self,*args, **kwargs): 150 """ 151 1. Run each available run type and record the run time and return value 152 2. Sort the run times from fastest to slowest 153 3. Compare each run type against each other, sorted by speed 154 155 :param args: args for the run method 156 :param kwargs: kwargs for the run method 157 :return: a list of tuples containing the run time, run type name and optionally the return values 158 :rtype: [[run_time, run_type_name, return_value], ...] 159 """ 160 161 # Create some lists to store runtimes and return values of run types 162 run_times = {} 163 returns = {} 164 165 # Run each run type and record the run time and return value 166 for run_type in self._run_types: 167 168 r = self._run(*args, run_type=run_type, **kwargs) 169 170 run_times[run_type] = self._last_time 171 172 if self.testing: # Store return values if testing 173 returns[run_type] = r 174 else: 175 returns[run_type] = None 176 177 # Sort run_times by value 178 speed_sort = [] 179 for run_type in sorted(run_times, key=run_times.get, reverse=False): 180 speed_sort.append( 181 ( 182 run_times[run_type], 183 run_type, 184 returns[run_type], 185 ) 186 ) 187 188 print(f"Fastest run type: {speed_sort[0][1]}") 189 print(f"Slowest run type: {speed_sort[-1][1]}") 190 191 # Compare each run type against each other, sorted by speed 192 for pair in combinations(speed_sort,2): 193 194 print(f"{pair[0][1]} is {pair[1][0]/pair[0][0]:.2f}x faster than {pair[1][1]}") 195 if self.testing: 196 if self._compare_runs(pair[0][2],pair[1][2]): 197 print(f"{pair[0][1]} and {pair[1][1]} have similar outputs!") 198 else: 199 print(f"WARNING: outputs of {pair[0][1]} and {pair[1][1]} don't match!") 200 201 return speed_sort
- Run each available run type and record the run time and return value
- Sort the run times from fastest to slowest
- Compare each run type against each other, sorted by speed
Parameters
- args: args for the run method
- kwargs: kwargs for the run method
Returns
a list of tuples containing the run time, run type name and optionally the return values
def
get_highest_divisor(self, size_, max_):
307 def get_highest_divisor(self, size_, max_): 308 """ 309 Returns the highest divisor of size_ that is still lower than max_ 310 """ 311 value = 1 312 for i in range(1, int(np.sqrt(size_)+1)): 313 if size_ % i==0: 314 if i*i != size_: 315 div2 = size_/i 316 317 if i < max_: 318 value = max(value, i) 319 if div2 < max_: 320 value = max(value, div2) 321 return int(value)
Returns the highest divisor of size_ that is still lower than max_
def
get_work_group(self, device, shape):
323 def get_work_group(self, device, shape): 324 """ 325 Calculates work group size for a given device and shape of global work space 326 """ 327 328 max_wg_dims = device.max_work_item_sizes[0:3] 329 max_glo_dims = device.max_work_group_size 330 331 three = self.get_highest_divisor(shape[2], max_wg_dims[2]) 332 max_two = max_glo_dims/three 333 two = self.get_highest_divisor(shape[1], max_two) 334 one = 1 335 #return (1,1,1) 336 return (one, two, three)
Calculates work group size for a given device and shape of global work space
def
run(self, *args, **kwargs):
345 def run(self, *args, **kwargs): 346 """ 347 Runs the function with the given args and kwargs 348 Should be overridden by the any class that inherits from this class 349 """ 350 return self._run(*args, **kwargs)
Runs the function with the given args and kwargs Should be overridden by the any class that inherits from this class